#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include <pthread.h>
#include <errno.h>
#include <time.h>
#include <stdbool.h>
#include "MQTTClient.h"
#include "shmem.h"
#include "msg_queue_peer.h"
#include "cJSON.h"
#include <sqlite3.h>

// ==================== 宏 ====================
#define N 32
#define SHM_SIZE 4096 // 共享内存大小
// MQTT连接参数
// #define ADDRESS "tcp://localhost:1883" // MQTT代理地址
#define CLIENTID "ExampleClientSub" // 客户端ID
#define PUB_TOPIC "up"              // 发布主题
#define SUB_TOPIC "down"            // 订阅主题
#define QOS 1                       // 服务质量等级
#define TIMEOUT 10000L              // 超时时间(毫秒)
// ==================== 定义 ====================
// mqtt服务器初始化配置
volatile MQTTClient_deliveryToken deliveredtoken;
MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient_deliveryToken token;
int rc;
int ch;
// 设备数据类型
typedef union
{
    int b_val;   // 布尔型
    int i_val;   // 整型
    float f_val; // 浮点型
} val_t;

// 共享内存节点结构
typedef struct
{
    int key;       // 节点唯一标识
    int type;      // 数据类型 (1:bool, 2:int, 3:float)
    int dev_type;  // 设备类型
    val_t old_val; // 旧值
    val_t new_val; // 新值
    int ret;       // 状态码
} std_node;

static struct shm_param para;     // 共享内存所需
static int *total;                // 指向共享内存中数据节点总个数
static std_node *node_arr;        // 指向共享内存中节点缓存数组头
static int stmsize;               // 点表中stm32的节点数量
static pthread_mutex_t shm_mutex; // 供共享内存使用的互斥锁
// static char configbuf[N * N * 2] = {0}; // 存放点表内容
static cJSON *reporttype; // 点表中数据上报模式
static cJSON *period;     // 点表中周期上报周期
static cJSON *root;       // 存储点表转成的json格式
static cJSON *report;     // 点表中上报属性report
static int change_i = 0;  // 变化上报中的标志位
sqlite3 *db;              //  数据库句柄
char *errmsg = NULL;      // 数据库错误信息指针

// 消息队列结构体
typedef struct
{
    long mtype;  // 控制的设备modbus=1,stm32=2
    int key;     // 要控制的设备key
    val_t value; // 要设置的值
    int type;    // 消息类型
} control_msg;

// 上报线程传递的参数
typedef struct
{
    pthread_t rpt_id; // 线程标识 (暂无用处)
    int tim;          // 周期上报的时间
} pth_report;

// ==================== 函数声明 ====================
void *handler_sql3(void *arg);                               // 数据库存储线程
int callback(void *arg, int num, char **value, char **name); // 查询数据库回调函数
void *handler_period(void *arg);                             // 周期上报线程函数
void *handler_change(void *arg);                             // 变化上报线程函数
void *report_refresh();                                      // 刷新上报函数
void delivered(void *context, MQTTClient_deliveryToken dt);
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message);
void connlost(void *context, char *cause);
// ==================== 主函数 ====================
int main(int argc, char const *argv[])
{
    // ---------- 1. 读取并转化配置点表 ----------

    FILE *file = fopen("./config/config.json", "r");
    if (file == NULL)
    {
        perror("fopen err\n");
        return -1;
    }
    char configbuf[N * N * 2] = {0};
    size_t len = fread(configbuf, sizeof(char), N * N * 2, file);
    if (len < 0)
    {
        printf("fread err\n");
        return -1;
    }
    fclose(file);
    // printf("%s\n", configbuf); // 查看点表能否完整读出
    //  把传入的字符串转成cJSON的结构
    root = cJSON_Parse(configbuf);
    if (root == NULL)
    {
        printf("cJSON_Parse err\n");
        return -1;
    }

    // ---------- 2. 建立mqtt服务器连接 ----------

    // 解析mqtt服务器配置
    cJSON *mqtt_server = cJSON_GetObjectItem(root, "mqtt_server");
    cJSON *addr = cJSON_GetObjectItem(mqtt_server, "addr");
    cJSON *port = cJSON_GetObjectItem(mqtt_server, "port");
    char ADDRESS[N];
    snprintf(ADDRESS, N, "tcp://%s:%d", addr->valuestring, port->valueint);

    // 建立mqtt服务器连接

    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL); // 创建客户端
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered); // 回调函数

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) // 连接
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
           "Press Q<Enter> to quit\n\n",
           SUB_TOPIC, CLIENTID, QOS);
    MQTTClient_subscribe(client, SUB_TOPIC, QOS); // 订阅QT客户端

    // ---------- 3. 共享内存初始设置 ----------

    pthread_mutex_init(&shm_mutex, NULL); // 初始化互斥锁

    // 获取共享内存中结构体节点数量
    cJSON *stm32 = cJSON_GetObjectItem(root, "stm32");
    if (stm32 == NULL)
    {
        printf("findstm32 err\n");
        return -1;
    }
    cJSON *modbus = cJSON_GetObjectItem(root, "modbus");
    if (modbus == NULL)
    {
        printf("findmodbus err\n");
        return -1;
    }
    cJSON *stmdata = cJSON_GetObjectItem(stm32, "data");
    if (stmdata == NULL)
    {
        printf("findstmdata err\n");
        return -1;
    }
    cJSON *modbusdata = cJSON_GetObjectItem(modbus, "data");
    if (modbusdata == NULL)
    {
        printf("findmodbusdata err\n");
        return -1;
    }
    stmsize = cJSON_GetArraySize(stmdata);
    int modbussize = cJSON_GetArraySize(modbusdata);
    printf("std_node_num = %d\n", stmsize + modbussize); // 共享内存中数据节点总个数 = stmsize + modbussize

    // 初始化共享内存
    int shmret = -1;
    shmret = shm_init(&para, "shm_test", SHM_SIZE);
    if (shmret < 0)
    {
        return -1;
    }
    void *node_p = shm_getaddr(&para);
    if (node_p == NULL)
    {
        printf("shm_getaddr err\n");
        return -1;
    }
    memset(node_p, 0, SHM_SIZE);
    total = (int *)node_p; // 指向共享内存头
    *total = stmsize + modbussize;
    node_arr = (std_node *)(node_p + sizeof(int)); // 指向共享内存结构体数组头

    printf("共享内存初始完成\n");
    // 向共享内存映射结构体节点数据
    for (int i = 0; i < *total; ++i)
    {
        if (i < stmsize)
        {
            cJSON *stmdata_i = cJSON_GetArrayItem(stmdata, i);
            cJSON *key = cJSON_GetObjectItem(stmdata_i, "key");
            cJSON *type = cJSON_GetObjectItem(stmdata_i, "type");
            node_arr[i].key = key->valueint;
            node_arr[i].type = type->valueint;
            node_arr[i].dev_type = 0;
            node_arr[i].ret = 0;
            // printf("%s = %d\n", key->string, node_arr[i].key);
            // printf("%s = %d\n", type->string, node_arr[i].type);
        }
        else
        {
            cJSON *modbusdata_i = cJSON_GetArrayItem(modbusdata, i - stmsize);
            cJSON *key = cJSON_GetObjectItem(modbusdata_i, "key");
            cJSON *type = cJSON_GetObjectItem(modbusdata_i, "type");
            node_arr[i].key = key->valueint;
            node_arr[i].type = type->valueint;
            node_arr[i].dev_type = 1;
            node_arr[i].ret = 0;
            // printf("%s = %d\n", key->string, node_arr[i].key);
            // printf("%s = %d\n", type->string, node_arr[i].type);
        }
    }
    printf("共享内存映射完成\n");
    // ---------- 4. 建立数据库 ---------------

    // 打开数据库
    if (sqlite3_open("./config/stu.db", &db) != SQLITE_OK)
    {
        fprintf(stderr, "open err:%s\n", sqlite3_errmsg(db));
        return -1;
    }
    printf("open okk\n");
    // 创建表
    if (sqlite3_exec(db, "create table if not exists stu(time long ,key int,val char)",
                     NULL, NULL, &errmsg) != 0)
    {
        fprintf(stderr, "create err:%s\n", sqlite3_errmsg(db));
        fprintf(stderr, "create err:%s\n", errmsg);
        return -1;
    }
    // 建立线程定时存储数据
    pthread_t sql3; // 数据库线程标识
    if (pthread_create(&sql3, NULL, handler_sql3, NULL) != 0)
    {
        perror("pthread_create err");
        return -1;
    }

    // ---------- 5. 数据上报策略 ----------

    // 解析上报模式，根据模式选择上报线程
    report = cJSON_GetObjectItem(root, "report");
    if (report == NULL)
    {
        printf("findreport err\n");
        return -1;
    }
    reporttype = cJSON_GetObjectItem(report, "type"); // 上报模式
    if (reporttype == NULL)
    {
        printf("findreporttype err\n");
        return -1;
    }
    period = cJSON_GetObjectItem(report, "period"); // 周期
    if (period == NULL)
    {
        printf("findreportperiod err\n");
        return -1;
    }

    pthread_t reportid;    // 上报线程pid
    pth_report reportargs; // 存储上报线程变量

    switch (reporttype->valueint)
    {
    case 0:
        // 刷新上报，先传输一次数据，然后等待指令
        report_refresh();
        break;
    case 1:
        // 变化上报
        // reportargs.rpt_id = reportid; // 暂时不用
        reportargs.tim = period->valueint;
        if (pthread_create(&reportid, NULL, handler_change, &reportargs) != 0)
        {
            perror("report_pthread_create err");
            return -1;
        }
        pthread_join(reportid, NULL);
        break;
    case 2:
        // 周期上报
        // reportargs.rpt_id = reportid; // 暂时不用
        reportargs.tim = period->valueint;
        if (pthread_create(&reportid, NULL, handler_period, &reportargs) != 0)
        {
            perror("report_pthread_create err");
            return -1;
        }
        pthread_join(reportid, NULL);
        break;
    default:
        printf("report_type err\n");
        break;
    }


    pthread_join(sql3, NULL); // 线程
    // mqtt断开
    MQTTClient_unsubscribe(client, SUB_TOPIC);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    sqlite3_close(db);        // 关闭数据库
    
    // shm_del(&para);                    // 删除共享内存
    pthread_mutex_destroy(&shm_mutex); // 销毁互斥锁
    cJSON_Delete(root);                // 删除json树
    return 0;
}

// ==================== 功能函数实现 ====================

// 数据库线程函数

void *handler_sql3(void *arg)
{
    char sqldata[128] = {0};
    time_t now;
    while (1)
    {
        sleep(30);
        pthread_mutex_lock(&shm_mutex);
        for (int i = 0; i < *total; ++i)
        {
            // key值 温度：4 湿度：16 光照: 6
            // 找到温度，并将数据存储
            if (node_arr[i].key == 4)
            {
                now = time(NULL); // 获取时间戳
                sprintf(sqldata, "insert into stu values(%ld,%d,'%f')",
                        now, node_arr[i].key, node_arr[i].new_val.f_val);
                // 向数据库插入数据
                if (sqlite3_exec(db, sqldata, NULL, NULL, &errmsg) != 0)
                {
                    fprintf(stderr, "insert err:%s\n", sqlite3_errmsg(db));
                    fprintf(stderr, "insert err:%s\n", errmsg);
                    return NULL;
                }
                memset(sqldata, 0, 128);
            }
            else if (node_arr[i].key == 6)
            {
                now = time(NULL); // 获取时间戳
                sprintf(sqldata, "insert into stu values(%ld,%d,'%d')",
                        now, node_arr[i].key, node_arr[i].new_val.i_val);
                // 向数据库插入数据
                if (sqlite3_exec(db, sqldata, NULL, NULL, &errmsg) != 0)
                {
                    fprintf(stderr, "insert err:%s\n", sqlite3_errmsg(db));
                    fprintf(stderr, "insert err:%s\n", errmsg);
                    return NULL;
                }
                memset(sqldata, 0, 128);
            }
            else if (node_arr[i].key == 16)
            {
                now = time(NULL); // 获取时间戳
                sprintf(sqldata, "insert into stu values(%ld,%d,'%f')",
                        now, node_arr[i].key, node_arr[i].new_val.f_val);
                // 向数据库插入数据
                if (sqlite3_exec(db, sqldata, NULL, NULL, &errmsg) != 0)
                {
                    fprintf(stderr, "insert err:%s\n", sqlite3_errmsg(db));
                    fprintf(stderr, "insert err:%s\n", errmsg);
                    return NULL;
                }
                memset(sqldata, 0, 128);
            }
        }
        pthread_mutex_unlock(&shm_mutex);
    }
}

// 数据库查询回调函数

int callback(void *arg, int num, char **value, char **name)
{
    cJSON *data = (cJSON *)arg;
    cJSON *item = cJSON_CreateObject();
    long tim_num = atol(value[0]);
    cJSON_AddItemToObject(item, "time", cJSON_CreateNumber(tim_num));
    cJSON_AddItemToObject(item, "val", cJSON_CreateString(value[2]));
    // 添加到data数组内
    cJSON_AddItemToArray(data, item);

    return 0; // 如果返回值不为0,报错
}

//---------- 上报线程函数 ----------

// 周期上报线程函数
void *handler_period(void *arg)
{
    // 创建JSon，插入数据
    pth_report *reportargs = (pth_report *)arg; // 接收参数
    cJSON *reportdata = cJSON_CreateObject();
    cJSON_AddItemToObject(reportdata, "type", cJSON_CreateNumber(1));   // 回复类型
    cJSON_AddItemToObject(reportdata, "result", cJSON_CreateNumber(0)); // 上报是否成功？
    while (1)
    {
        sleep(reportargs->tim); // 周期
        // 创建数据存放节点
        cJSON *data = cJSON_CreateArray();
        if (data == NULL)
        {
            printf("reporrtdata err\n");
            return NULL;
        }
        cJSON_AddItemToObject(reportdata, "data", data);
        printf("add_report_data begin\n");
        // 申请互斥锁
        pthread_mutex_lock(&shm_mutex);
        // 遍历插入数据
        for (int i = 0; i < *total; ++i)
        {
            cJSON *partdata = cJSON_CreateObject();

            cJSON_AddItemToObject(partdata, "key", cJSON_CreateNumber(node_arr[i].key));
            char value_str[N]; // 存数据字符串
            // 将数据转为字符串
            switch (node_arr[i].type)
            {
            case 1:
                snprintf(value_str, sizeof(value_str), "%d", node_arr[i].new_val.b_val);
                break;
            case 2:
                snprintf(value_str, sizeof(value_str), "%d", node_arr[i].new_val.i_val);
                break;
            case 3:
                snprintf(value_str, sizeof(value_str), "%f", node_arr[i].new_val.f_val);
                break;
            default:
                printf("data report type err\n");
            }
            // 添加具体数据
            cJSON_AddItemToObject(partdata, "val", cJSON_CreateString(value_str));
            cJSON_AddItemToArray(data, partdata);
            memset(value_str, 0, N);
        }
        pthread_mutex_unlock(&shm_mutex);          // 释放互斥锁
        char *reportbuf = cJSON_Print(reportdata); // 转字符串
        // mqtt发送
        // printf("%s", reportbuf);

        pubmsg.payload = reportbuf;
        pubmsg.payloadlen = (int)strlen(reportbuf);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        MQTTClient_publishMessage(client, PUB_TOPIC, &pubmsg, &token);
        printf("report_period\n");
        // printf("Waiting for up to %d seconds for publication of data\n"
        //        "on topic %s for client with ClientID: %s\n",
            //    (int)(TIMEOUT / 1000), PUB_TOPIC, CLIENTID);
        rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
        // printf("Message with delivery token %d delivered\n", token);

        free(reportbuf);
        cJSON_DeleteItemFromObject(reportdata, "data");
    }
    pthread_exit(NULL);
}

// 变化上报线程函数

void *handler_change(void *arg)
{
    // 先刷新上报一次
    report_refresh();
    // 创建JSon，插入数据
    pth_report *reportargs = (pth_report *)arg; // 接收参数
    cJSON *reportdata = cJSON_CreateObject();
    cJSON_AddItemToObject(reportdata, "type", cJSON_CreateNumber(1));   // 回复类型
    cJSON_AddItemToObject(reportdata, "result", cJSON_CreateNumber(0)); // 上报是否成功？
    while (1)
    {
        for (int i = 0; i < *total; ++i)
        {
            pthread_mutex_lock(&shm_mutex); // 申请互斥锁
            if (node_arr[i].type == 1 && node_arr[i].new_val.b_val != node_arr[i].old_val.b_val)
            {
                change_i = 1;
            }
            else if (node_arr[i].type == 2 && node_arr[i].new_val.i_val != node_arr[i].old_val.i_val)
            {
                change_i = 2;
            }
            else if (node_arr[i].type == 3 && node_arr[i].new_val.f_val != node_arr[i].old_val.f_val)
            {
                change_i = 3;
            }

            if (change_i)
            {
                change_i = 0;                      // 标志位置零
                cJSON *data = cJSON_CreateArray(); // data数组总节点
                if (data == NULL)
                {
                    printf("reporrtdata err\n");
                    return NULL;
                }
                cJSON_AddItemToObject(reportdata, "data", data);
                printf("add_report_data begin\n");

                cJSON *partdata = cJSON_CreateObject(); // 创建数组元素
                // 添加key值到数组
                cJSON_AddItemToObject(partdata, "key", cJSON_CreateNumber(node_arr[i].key));
                char value_str[N]; // 存数据字符串
                // 将数据转为字符串
                switch (node_arr[i].type)
                {
                case 1:
                    snprintf(value_str, sizeof(value_str), "%d", node_arr[i].new_val.b_val);
                    break;
                case 2:
                    snprintf(value_str, sizeof(value_str), "%d", node_arr[i].new_val.i_val);
                    break;
                case 3:
                    snprintf(value_str, sizeof(value_str), "%f", node_arr[i].new_val.f_val);
                    break;
                default:
                    printf("data report type err\n");
                }
                // 添加具体数据
                cJSON_AddItemToObject(partdata, "val", cJSON_CreateString(value_str));
                cJSON_AddItemToArray(data, partdata); // 将数组元素加入到数组中
                memset(value_str, 0, N);

                char *reportbuf = cJSON_Print(reportdata); // 转字符串
                // mqtt发送
                // printf("%s", reportbuf);

                pubmsg.payload = reportbuf;
                pubmsg.payloadlen = (int)strlen(reportbuf);
                pubmsg.qos = QOS;
                pubmsg.retained = 0;
                MQTTClient_publishMessage(client, PUB_TOPIC, &pubmsg, &token);
                printf("report_change\n");
                // printf("Waiting for up to %d seconds for publication of data\n"
                //        "on topic %s for client with ClientID: %s\n",
                //        (int)(TIMEOUT / 1000), PUB_TOPIC, CLIENTID);
                rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
                // printf("Message with delivery token %d delivered\n", token);

                free(reportbuf);
                cJSON_DeleteItemFromObject(reportdata, "data");
            }
            pthread_mutex_unlock(&shm_mutex); // 销毁互斥锁
        }
    }
}

// 刷新上报函数

void *report_refresh()
{
    // 创建JSon，插入数据
    cJSON *reportdata = cJSON_CreateObject();                           // 最外层{}
    cJSON_AddItemToObject(reportdata, "type", cJSON_CreateNumber(1));   // 回复类型
    cJSON_AddItemToObject(reportdata, "result", cJSON_CreateNumber(0)); // 上报是否成功？

    // 创建数据存放节点
    cJSON *data = cJSON_CreateArray();
    if (data == NULL)
    {
        printf("reporrtdata err\n");
        return NULL;
    }
    cJSON_AddItemToObject(reportdata, "data", data);
    printf("add_report_data begin\n");

    // 申请互斥锁
    pthread_mutex_lock(&shm_mutex);
    // 遍历插入数据
    for (int i = 0; i < *total; ++i)
    {
        cJSON *partdata = cJSON_CreateObject();
        cJSON_AddItemToObject(partdata, "key", cJSON_CreateNumber(node_arr[i].key));
        char value_str[N]; // 存数据字符串
        // 将数据转为字符串
        switch (node_arr[i].type)
        {
        case 1:
            snprintf(value_str, sizeof(value_str), "%d", node_arr[i].new_val.b_val);
            break;
        case 2:
            snprintf(value_str, sizeof(value_str), "%d", node_arr[i].new_val.i_val);
            break;
        case 3:
            snprintf(value_str, sizeof(value_str), "%f", node_arr[i].new_val.f_val);
            break;
        default:
            printf("data report type err\n");
        }
        // 添加具体数据
        cJSON_AddItemToObject(partdata, "val", cJSON_CreateString(value_str));
        cJSON_AddItemToArray(data, partdata);
        memset(value_str, 0, N);
    }
    char *reportbuf = cJSON_Print(reportdata); // 转字符串
    pthread_mutex_unlock(&shm_mutex);          // 释放互斥锁

    // mqtt发送
    pubmsg.payload = reportbuf;
    pubmsg.payloadlen = (int)strlen(reportbuf);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client, PUB_TOPIC, &pubmsg, &token);
    printf("report_refresh\n");
    // printf("Waiting for up to %d seconds for publication of data\n"
    //        "on topic %s for client with ClientID: %s\n",
    //        (int)(TIMEOUT / 1000), PUB_TOPIC, CLIENTID);

    free(reportbuf);
    cJSON_Delete(reportdata);
}

//---------- mqtt相关函数 ----------

// 消息投递确认回调
void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

// MQTT消息到达回调
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{

    // 解析命令，响应，转发
    cJSON *msg_json = cJSON_Parse(message->payload);         // 将消息转为json格式
    cJSON *msg_type = cJSON_GetObjectItem(msg_json, "type"); // 指令类型
    if (msg_type->valueint == 1)                             // 设备采集指令
    {
        report_refresh(); // 刷新上报
    }
    else if (msg_type->valueint == 2) // 设备控制指令
    {
        cJSON *msg_data = cJSON_GetObjectItem(msg_json, "data"); // 控制指令
        cJSON *msg_key = cJSON_GetObjectItem(msg_data, "key");   // 拿key值
        cJSON *msg_val = cJSON_GetObjectItem(msg_data, "val");   // 拿val值
        control_msg msg_send;
        // 接收val值
        int control_vali;   // bool和整形
        float control_valf; // 浮点型
        // 申请互斥锁
        pthread_mutex_lock(&shm_mutex);
        // 判断发给哪个设备，将字符串的val值转为相应的数据类型
        for (int i = 0; i < *total; ++i)
        {
            if (i < stmsize)
            {
                if (node_arr[i].key == msg_key->valueint)
                {
                    msg_send.mtype = 2;
                    msg_send.key = msg_key->valueint;
                    msg_send.type = node_arr[i].type;
                    // 字符串转共用体具体数据
                    switch (node_arr[i].type)
                    {
                    case 1:
                        control_vali = atoi(msg_val->valuestring);
                        msg_send.value.b_val = control_vali;
                        break;
                    case 2:
                        control_vali = atoi(msg_val->valuestring);
                        msg_send.value.i_val = control_vali;
                        break;
                    case 3:
                        control_valf = atof(msg_val->valuestring);
                        msg_send.value.f_val = control_valf;
                        break;
                    }
                }
            }
            else
            {
                if (node_arr[i].key == msg_key->valueint)
                {
                    msg_send.mtype = 1;
                    msg_send.key = msg_key->valueint;
                    msg_send.type = node_arr[i].type;
                    switch (node_arr[i].type)
                    {
                    case 1:
                        control_vali = atoi(msg_val->valuestring);
                        msg_send.value.b_val = control_vali;
                        break;
                    case 2:
                        control_vali = atoi(msg_val->valuestring);
                        msg_send.value.i_val = control_vali;
                        break;
                    case 3:
                        control_valf = atof(msg_val->valuestring);
                        msg_send.value.f_val = control_valf;
                        break;
                    }
                }
            }
        }
        pthread_mutex_unlock(&shm_mutex); // 释放互斥锁
        if (msg_queue_send("device_control", &msg_send, sizeof(msg_send), 0) < 0)
        {
            printf("msg_queue_send error\n");
            return -1;
        }
        printf("msg_send.mtype=%ld,msg_send.value=%d\n", msg_send.mtype, msg_send.value.b_val);
    }
    else if (msg_type->valueint == 3) // 模式修改
    {
        cJSON *msg_data = cJSON_GetObjectItem(msg_json, "data");      // 控制指令
        cJSON *msg_data_type = cJSON_GetObjectItem(msg_data, "type"); // 拿type值

        // 判断具体情况，选择修改配置文件
        if (msg_data_type->valueint == reporttype->valueint && reporttype->valueint != 2)
        {
            printf("当前上报模式与目标相同,无需修改\n"); // 要修改的模式与当前模式相同（除周期上报模式外）
        }
        else if (msg_data_type->valueint == 2)
        {
            cJSON *msg_period = cJSON_GetObjectItem(msg_data, "period");
            if (msg_period->valueint == period->valueint &&
                msg_data_type->valueint == reporttype->valueint)
            {
                printf("当前上报模式与目标相同,无需修改\n"); // 周期上报模式且上报周期相同
            }
            else
            {
                // 替换report上的period节点
                cJSON_ReplaceItemInObject(report, "type", cJSON_CreateNumber(msg_data_type->valueint));
                cJSON_ReplaceItemInObject(report, "period", cJSON_CreateNumber(msg_period->valueint));
                char *config = cJSON_Print(root); // json转字符串
                // 打开配置文件将字符串写入
                FILE *file = fopen("./config/config.json", "w");
                if (file == NULL)
                {
                    perror("fopen err\n");
                    free(config);
                    return -1;
                }
                size_t len = fwrite(config, sizeof(char), strlen(config), file);
                if (len < 0)
                {
                    printf("fwrite err\n");
                    return -1;
                }
                free(config);
                fclose(file);
            }
        }
        else
        {
            // 替换report上的type节点
            cJSON_ReplaceItemInObject(report, "type", cJSON_CreateNumber(msg_data_type->valueint));
            char *config = cJSON_Print(root); // json转字符串
            // 打开配置文件将字符串写入
            FILE *file = fopen("./config/config.json", "w");
            if (file == NULL)
            {
                perror("fopen err\n");
                free(config);
                return -1;
            }
            size_t len = fwrite(config, sizeof(char), strlen(config), file);
            if (len < 0)
            {
                printf("fwrite err\n");
                free(config);
                fclose(file);
                return -1;
            }
            free(config);
            fclose(file);
        }
    }
    else if (msg_type->valueint == 4)
    {
        // 读取指令
        cJSON *msg_data = cJSON_GetObjectItem(msg_json, "data");   // 历史数据data
        cJSON *msg_key = cJSON_GetObjectItem(msg_data, "key");     // 拿key值
        cJSON *msg_limit = cJSON_GetObjectItem(msg_data, "limit"); // 找limit
        cJSON *tim_min = cJSON_GetArrayItem(msg_limit, 0);         // 拿时间戳
        cJSON *tim_max = cJSON_GetArrayItem(msg_limit, 1);         // 拿时间戳

        // 创建回复json
        cJSON *history = cJSON_CreateObject();                                        // 最外层{}
        cJSON_AddItemToObject(history, "type", cJSON_CreateNumber(4));                // 回复类型
        cJSON_AddItemToObject(history, "result", cJSON_CreateNumber(0));              // 上报是否成功？
        cJSON_AddItemToObject(history, "key", cJSON_CreateNumber(msg_key->valueint)); // key值

        // 创建数据存放节点
        cJSON *data = cJSON_CreateArray();
        if (data == NULL)
        {
            printf("reporrtdata err\n");
            return -1;
        }
        cJSON_AddItemToObject(history, "data", data);

        // 从数据库拿数据并通过回调函数加入树
        char sqlbuf[128];
        sprintf(sqlbuf, "select * from stu where key=%d and time>%f and time<%f",
                msg_key->valueint, tim_min->valuedouble, tim_max->valuedouble); // 数据库指令
        if (sqlite3_exec(db, sqlbuf, callback, data, &errmsg) != 0)
        {
            fprintf(stderr, "select err:%s\n", sqlite3_errmsg(db));
            fprintf(stderr, "select err:%s\n", errmsg);
            return -1;
        }

        char *reportbuf = cJSON_Print(history); // 转字符串
        // mqtt发送
        pubmsg.payload = reportbuf;
        pubmsg.payloadlen = (int)strlen(reportbuf);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        MQTTClient_publishMessage(client, PUB_TOPIC, &pubmsg, &token);
        printf("history send\n");

        free(reportbuf);
        cJSON_Delete(history);
    }

    cJSON_Delete(msg_json);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

// 连接丢失回调
void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}